package cc.rboot.topic.processor;

import cc.rboot._core.holder.AppContextHolder;
import cc.rboot.topic.core.AbstractMessageProcessor;
import cc.rboot.topic.core.IMessageConsumer;
import org.redisson.api.RSet;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;

import java.util.Set;

public class SubscribeMessageProcessor<T> extends AbstractMessageProcessor<T> {

    @Override
    public void register(IMessageConsumer<T> consumer) {
        boundConsumer(consumer.topic().name()).add(consumer.name());

        RTopic topic = boundTopic(consumer.name());
        topic.addListener(Object.class, (CharSequence charSequence, Object o) -> {
            boolean ifSuccess = false;
            try {
                ifSuccess = consumer.onMessage((T) o);
            } catch (Exception ex) {
                // Do nothing
            }
            if (!ifSuccess) {
                try {
                    topic.publish(o);
                } catch (Exception ex) {
                    // 打印下日志即可，一般消费者都是回调，出错了也没有报错的对象
                    System.out.println(ex.getMessage());
                }
            }
        });
    }

    @Override
    public void send(T message) throws Exception {
        Set<String> boundConsumerSet = boundConsumerSet();
        if (boundConsumerSet.size() == 0) {
            throw new Exception("消费者未注册，无法发送消息");
        }
        for (String consumerName : boundConsumerSet()) {
            boundTopic(consumerName).publish(message);
        }
    }

    private Set<String> boundConsumerSet() {
        return boundConsumer(super.getTopic()).readAll();
    }

    private RTopic boundTopic(String consumerName) {
        String key = "TOPIC:MESSAGE:" + this.getTopic() + ":" + consumerName;
        return getClient().getTopic(key);
    }

    private RSet<String> boundConsumer(String topicName) {
        return getClient().getSet("TOPIC:" + topicName);
    }


    private RedissonClient getClient() {
        return AppContextHolder.getBean("redissonClient");
    }

}
